Skip to content

03 使用Python开发MCP Server

Python MCP SDK提供了FastMCP封装,可以通过装饰器快速定义Tools、Resources和Prompts。对于大多数业务场景,可以先用FastMCP完成Server原型,再根据安全、部署和权限需求逐步增强。

本文实现一个文档查询MCP Server,包含:

  • 一个搜索工具:search_docs
  • 一个文档资源:docs://article/{slug}
  • 一个提示词模板:summarize_article

一、项目初始化

创建项目目录:

bash
uv init docs-mcp-server
cd docs-mcp-server
uv venv
source .venv/bin/activate

安装依赖:

bash
uv add "mcp[cli]"

创建入口文件:

bash
touch server.py

项目结构如下:

text
docs-mcp-server/
├── server.py
└── pyproject.toml

二、创建FastMCP实例

server.py中初始化MCP Server:

python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-server")

FastMCP("docs-server")中的名称用于标识当前Server。Host或调试工具连接Server时,可以通过该名称区分不同服务。

三、准备示例数据

为了让示例可以独立运行,先用内存字典模拟文档库:

python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-server")

ARTICLES = {
    "docker-intro": {
        "title": "Docker基础入门",
        "content": "Docker用于把应用和运行环境打包成镜像,减少环境差异。",
        "tags": ["docker", "container"],
    },
    "mcp-architecture": {
        "title": "MCP基础与架构",
        "content": "MCP通过Host、Client、Server连接AI应用和外部系统。",
        "tags": ["mcp", "agent"],
    },
}

真实项目中,ARTICLES可以替换成数据库查询、向量检索、文件读取或内部HTTP API。

四、暴露Tool

工具用于执行动作。这里定义一个search_docs工具,用于按关键词搜索文章:

python
@mcp.tool()
def search_docs(query: str) -> list[dict]:
    """Search articles by keyword.

    Args:
        query: Keyword used to search article title, content, or tags.
    """
    keyword = query.lower()
    results = []

    for slug, article in ARTICLES.items():
        haystack = " ".join(
            [article["title"], article["content"], *article["tags"]]
        ).lower()
        if keyword in haystack:
            results.append(
                {
                    "slug": slug,
                    "title": article["title"],
                    "summary": article["content"],
                }
            )

    return results

FastMCP会根据函数签名和docstring生成工具定义,包括工具名称、描述和参数schema。模型能否正确使用工具,很大程度取决于函数命名、参数类型和docstring是否清晰。

五、暴露Resource

Resource用于提供只读上下文。这里定义一个按slug读取文章的资源:

python
@mcp.resource("docs://article/{slug}")
def read_article(slug: str) -> str:
    """Read an article by slug."""
    article = ARTICLES.get(slug)
    if not article:
        raise ValueError(f"Article not found: {slug}")

    return f"# {article['title']}\n\n{article['content']}"

资源URI是Host识别和读取资源的入口。docs://article/{slug}表示这是一个带参数的资源模板。

Resource适合承载上下文数据,例如文档正文、数据库schema、配置说明等。与Tool不同,Resource通常不应该修改外部系统状态。

六、暴露Prompt

Prompt用于提供可复用的任务模板。这里定义一个文章总结模板:

python
@mcp.prompt()
def summarize_article(slug: str) -> str:
    """Create a prompt for summarizing an article."""
    return (
        "请读取资源 docs://article/{slug},并按以下结构总结:\n"
        "1. 核心主题\n"
        "2. 关键概念\n"
        "3. 适用场景\n"
        "4. 注意事项\n"
    ).format(slug=slug)

Prompt适合沉淀标准任务流程。例如代码审查、故障排查、文章总结、需求分析等。

七、启动Server

本地MCP Server通常使用stdio传输。添加启动入口:

python
def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

完整文件如下:

python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-server")

ARTICLES = {
    "docker-intro": {
        "title": "Docker基础入门",
        "content": "Docker用于把应用和运行环境打包成镜像,减少环境差异。",
        "tags": ["docker", "container"],
    },
    "mcp-architecture": {
        "title": "MCP基础与架构",
        "content": "MCP通过Host、Client、Server连接AI应用和外部系统。",
        "tags": ["mcp", "agent"],
    },
}


@mcp.tool()
def search_docs(query: str) -> list[dict]:
    """Search articles by keyword.

    Args:
        query: Keyword used to search article title, content, or tags.
    """
    keyword = query.lower()
    results = []

    for slug, article in ARTICLES.items():
        haystack = " ".join(
            [article["title"], article["content"], *article["tags"]]
        ).lower()
        if keyword in haystack:
            results.append(
                {
                    "slug": slug,
                    "title": article["title"],
                    "summary": article["content"],
                }
            )

    return results


@mcp.resource("docs://article/{slug}")
def read_article(slug: str) -> str:
    """Read an article by slug."""
    article = ARTICLES.get(slug)
    if not article:
        raise ValueError(f"Article not found: {slug}")

    return f"# {article['title']}\n\n{article['content']}"


@mcp.prompt()
def summarize_article(slug: str) -> str:
    """Create a prompt for summarizing an article."""
    return (
        "请读取资源 docs://article/{slug},并按以下结构总结:\n"
        "1. 核心主题\n"
        "2. 关键概念\n"
        "3. 适用场景\n"
        "4. 注意事项\n"
    ).format(slug=slug)


def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

启动Server:

bash
uv run server.py

stdio模式下,Server会等待MCP Client通过标准输入输出发送JSON-RPC消息。直接运行时终端可能不会出现普通Web服务那样的监听地址。

八、stdio日志注意事项

使用stdio传输时,Server的stdout用于发送MCP协议消息。因此,业务日志不能写到stdout,否则会破坏JSON-RPC通信。

错误示例:

python
print("server started")

推荐写法:

python
import sys

print("server started", file=sys.stderr)

也可以使用日志库,并确保日志输出到stderr或文件。

python
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

logger.info("server started")

对于HTTP传输,标准输出不会干扰HTTP响应,但仍建议使用结构化日志,便于排查工具调用和权限问题。

九、开发建议

开发MCP Server时,可以按以下顺序推进:

  1. 先用内存数据或本地文件完成最小Server。
  2. @mcp.tool()暴露最核心的查询类工具。
  3. 再补充Resource,让Host可以读取上下文。
  4. 将重复任务沉淀为Prompt。
  5. 使用MCP Inspector测试工具列表、资源读取和提示词模板。
  6. 接入真实系统前补充参数校验、权限控制和日志。

十、总结

使用Python开发MCP Server的关键步骤如下:

步骤说明
初始化Server创建FastMCP实例
暴露Tool使用@mcp.tool()定义可执行函数
暴露Resource使用@mcp.resource()定义可读取上下文
暴露Prompt使用@mcp.prompt()定义提示词模板
启动Server本地开发通常使用stdio传输
调试使用Inspector检查能力发现和调用结果

完成最小Server后,就可以进入传输方式、客户端接入和部署配置。